package epoch.game.core.actor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;

import epoch.game.core.action.IAction;
import epoch.game.core.config.Config;
import epoch.game.core.distr.Node;
import epoch.game.core.distr.NodeMessage;
import epoch.game.core.distr.call.Call;
import epoch.game.core.distr.call.CallContext;
import epoch.game.core.distr.call.CallResult;
import epoch.game.core.distr.call.CallReturn;
import epoch.game.core.distr.service.Service;
import epoch.game.core.function.FuncUtil;
import epoch.game.core.message.SysMessageIds;
import epoch.game.core.serialize.StreamWriter;
import epoch.game.core.stat.StatisticsRPC;
import epoch.game.core.util.BeanUtil;
import epoch.game.core.util.Utils;

public class Actor {
	ASystem AS;
	int id;
	String name;
	private Queue<AMessage> mq;
	private SpinLock lock;
	private boolean stop;
	volatile boolean interrupt;
	
	//回调函数及回调上下文参数
	private HashMap<Long,CallResult> callbackMap = new HashMap<Long, CallResult>();
	private long lastTimeout;
	
	public Actor() {
		this.AS = null;
		this.id = 0;
		this.mq = new LinkedList<AMessage>();
		this.lock = new SpinLock();
		this.stop = false;
		this.interrupt = false;
	}
	
	void stop(boolean force) {
		int size;
		lock.lock();
		stop = true;
		size = mq.size();
		lock.unlock();
		if (force) {
			if (size > 0) {
				Log.info(this, "actor " + id + " stoped with mqSize " + size);
			}
		} else {
			for (int i = 0; i < size; ++i) {
				AMessage msg;
				lock.lock();
				msg = mq.poll();
				lock.unlock();
				if (msg == null) {
					break;
				}
				handle(msg);
			}
		}
	}
	void handle(AMessage msg) {
		try {
			switch (msg.type) {
			case ACTION:
				IAction action = (IAction)msg.data;
				action.action();
				break;
			case CALL:
				Call call = (Call)msg.data;
				handleCall(call);
				break;
			case TIMEOUT:
				ATimer atimer = (ATimer)msg.data;
				if (atimer.tick()) {
					regTimer(atimer);
				}
				break;
			case RET:
				onReturn((CallReturn)msg.data);
				break;
			case INIT:
				regCbTimeout();
				init();
				break;
			default:
				message(msg);
				break;
			}
		} catch (Throwable e) {
			Log.error(this, "Actor dispatchMessage error: ", e);
		}
	}
	int dispatchMessage() {
		int n;
		lock.lock();
		if (stop) {
			lock.unlock();
			return -1;
		}
		n = mq.size();
		lock.unlock();
		
		int r = 0;
		for (int i=0; i<n; ++i) {
			AMessage msg;
			lock.lock();
			msg = mq.poll();
			lock.unlock();
			if (msg == null) {
				break;
			}
			handle(msg);
			r++;
			if (interrupt) {
				interrupt = false;
				Log.error(this, "Actor dispatchMessage interrupted");
				break;
			}
		}
		return r;
	}
	public boolean send(int source, AMessage.Type type, Object data) {
		AMessage msg = new AMessage(source, type, data);
		boolean ok;
		lock.lock();
		ok = mq.offer(msg);
		lock.unlock();
		return ok;
	}
	public boolean send(IAction action) {
		int source = AContext.selfId();
		return send(source, AMessage.Type.ACTION, action);
	}
	public boolean call(Call call) {
		int source = AContext.selfId();
		return send(source, AMessage.Type.CALL, call);
	}
	public boolean regTimer(ATimer timer) {
		return AS.timerActor.send(id, AMessage.Type.REGTIMER, timer);
	}
	public int mqSize() {
		int size;
		lock.lock();
		size = mq.size();
		lock.unlock();
		return size;
	}
	public int actorId() {
		return id;
	}
	public String actorName() {
		return name;
	}
	protected void init() {
		// no handle init
	}
	protected void message(AMessage msg) {
		// no handle message
	}
	private void checkCbTimeout() {
		lastTimeout = AS.now();
		List<CallResult> rm = null;
		for(CallResult cr : callbackMap.values()){
			if(lastTimeout >= cr.timeOut){
				if (rm == null) {
					rm = new ArrayList<>();
				}
				rm.add(cr);
			}
		}
		if (rm != null) {
			for(CallResult cr : rm){
				callbackMap.remove(cr.callId);
				if(cr.handleTimeOut){
					//执行超时回调
					if(cr.context!=null){
						cr.context.markTimeOut();
						try {
							FuncUtil.apply(cr.callbackMethod, cr.context, null);
						} catch (Exception ex) {
							Log.error(this, "CallBack time out handle ERROR!", ex);
						}
					}
				}
			}
		}
	}
	private void regCbTimeout() {
		lastTimeout = AS.now();
		regTimer(new ATimer(1000) {
			protected void doAction() {
				checkCbTimeout();
			}
		});
	}
	
	// 临时兼容
	private long callId = 1;
	private long currentCallbackId = 0;

	// 当前正在处理的call,生命周期只存在call处理期间
	public Call currentCall;
	// 当前需要处理的回调
	public CallReturn callReturn;
	//当前的返回值
	public Object returns;
	private long waitSyncCallId=0;
	
	public long currentCallbackId() {
		return currentCallbackId;
	}
	public long getAndResetCallbackId() {
		long old = currentCallbackId;
		currentCallbackId = 0;
		return old;
	}
	public Call currentCall() {
		return currentCall;
	}
	public void listen(Object callback, CallContext context, boolean handleTimeOut) {
		CallResult result = new CallResult();
		result.callId = callId++;
		result.timeOut = AS.now() + 60000;
		result.callbackMethod = callback;
		if(context!=null){
			result.context = context;
			result.handleTimeOut=handleTimeOut;
			result.context.setTimeOut(false);
		}
		if(Config.STAT_ENABLE){
			result.callerInfo = getCallResultCallerInfo();
			result.listenTime=System.currentTimeMillis();
		}
		callbackMap.put(result.callId, result);
		currentCallbackId = result.callId;
	}
	private String getCallResultCallerInfo() {
		return BeanUtil.getStack();
	}

	private void onReturn(CallReturn ret) {
		long callbackId = ret.id;
		Object returns = ret.getReturns();
		
		//添加到线程进行处理
		CallResult callResult = callbackMap.remove(callbackId);
		if(callResult==null){
			//丢失，不处理,记录日志
			return;
		}
		try {
			//执行回调
			if(callResult.context==null){
				FuncUtil.apply(callResult.callbackMethod, returns);
			}else{
				FuncUtil.apply(callResult.callbackMethod, callResult.context, returns);
			}		
		} catch (Exception ex) {
			Log.error(this, "处理回调出错,callerInfo="+callResult.callerInfo+",ex="+ex, ex);
		}
	}
	private void handleCall(Call call) {
		try{
			Service service = call.service;
			currentCall = call;
			Object method = service.getMethod(call.methodKey);
			Object[] p = call.methodParam;
			if(call.callbackId>0){
				callReturn = call.createCallReturn();
			}else{
				callReturn = null;
			}
			//开启统计
			long start = Config.STAT_ENABLE ? System.nanoTime() : 0;
			FuncUtil.apply(method, p);
			doReturn();
			if(start>0){
				StatisticsRPC.rpc(Utils.createStr("{}${}", service.getClass().toString(), call.methodKey), System.nanoTime()-start);
			}
		}catch(Exception ex){
			Log.error(this, "处理远程Call出错,callerInfo="+call.callerInfo+",ex="+ex, ex);
		}
	}
	public void doReturn(){	
		doReturn(callReturn, returns);	
		currentCall = null;
		returns=null;
		callReturn = null;
	}
	public void doReturn(CallReturn callReturn, Object returns){	
		if(callReturn==null){
			return;
		}
		try {
			callReturn.setReturns(returns);
			ASystem AS = AContext.SYSTEM.get();
			if (AS != null) {
				Node node = AS.node;
				if (node == null ||
					callReturn.nodeId == AS.node.getId()) {
					AS.onReturn(callReturn);
				} else {
					StreamWriter writer = new StreamWriter();
					callReturn.writeTo(writer);
					byte[] calldata = writer.array();
					NodeMessage nodeMsg = new NodeMessage();
					nodeMsg.setId(SysMessageIds.rpc_return);
					nodeMsg.setMsgData(calldata);
					node.sendNodeMsg(callReturn.nodeId, nodeMsg);
				}
			} else {
				Log.error(this, "Actor doReturn not in ASystem, just skip");
			}
		} catch (Exception ex) {
			Log.error(this, "Actor doReturn error: ", ex);
		}
	}
	public void listenSync(){
		if(waitSyncCallId!=0){
			//上一个同步调用未完成，不能嵌套执行同步调用
			throw new RuntimeException("current sync call not finish!");
		}
		waitSyncCallId = currentCallbackId = callId++;
	}
	public <T> T waitFor(long timeout){
		if(waitSyncCallId==0){
			return null;
		}
		CallReturn callRet = null;
		long timeOutTime = System.currentTimeMillis()+timeout;
		while (true) {
			AMessage msg;
			lock.lock();
			msg = mq.poll();
			lock.unlock();
			if (msg != null) {
				if (msg.type == AMessage.Type.RET) {
					CallReturn cr = (CallReturn)msg.data;
					if (cr.id == waitSyncCallId) {
						callRet = cr;
						break;
					}
				}
				handle(msg);
			} else {
				if (System.currentTimeMillis() > timeOutTime) {
					break;
				}
				try {
					Thread.sleep(10);
				} catch (InterruptedException e){
				}
			}
		}
		waitSyncCallId=0;
		if(callRet==null){
			return null;
		}
		return callRet.getReturns();
	}
}